%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.

-module(emqx_plugin_rabbitmq).

-include_lib("emqx/include/emqx.hrl").
-include("../include/emqx_plugin_rabbitmq.hrl").

-export([ load/1
        , unload/0
        ]).

%% Hooks functions
-export([ 
        on_client_connected/4
        , on_client_disconnected/3
        , on_message_publish/2
        ]).

-import(emqx_plugin_rabbitmq_cli, [ensure_exchange/1, publish/3]).
-import(bson_binary, [put_document/1]).

%% Called when the plugin application start
load(Env) ->

    {ok, Host} = application:get_env(?APP, hook_rabbitmq_host),
    {ok, Port} = application:get_env(?APP, hook_rabbitmq_port),
    {ok, PoolSize} = application:get_env(?APP, hook_rabbitmq_poolSize),
    {ok, UserName} = application:get_env(?APP, hook_rabbitmq_username),
    {ok, Password} = application:get_env(?APP, hook_rabbitmq_password),
    {ok, ExchangeName} = application:get_env(?APP, hook_rabbitmq_exchangeName),
    {ok, Prefer_ipv6} = application:get_env(?APP, hook_rabbitmq_prefer_ipv6),

    AmqpOpts = [
          {host, Host},
          {port, Port},
          {username, UserName},
          {password, Password},
          {pool_size, PoolSize}],

    ecpool:start_pool(rabbitmq_pool, emqx_plugin_rabbitmq_cli, AmqpOpts),

    emqx:hook('client.connected', fun ?MODULE:on_client_connected/4, [Env]),
    emqx:hook('client.disconnected', fun ?MODULE:on_client_disconnected/3, [Env]),
    emqx:hook('message.publish', fun ?MODULE:on_message_publish/2, [Env]).


on_client_connected(ClientInfo = #{client_id := ClientId}, ConnAck, ConnAttrs, _Env) ->
    {ok, ExchangeName} = application:get_env(?APP, hook_rabbitmq_exchangeName),
    Payload = [{action, <<"connected">>},
        {clientId, ClientId},
        {username, maps:get(username, ClientInfo)},
        {timestamp, calendar:datetime_to_gregorian_seconds(calendar:universal_time())-719528*24*3600}],
    emqx_plugin_rabbitmq_cli:publish(ExchangeName, Payload, <<"emqx_message">>),
    io:format("Client(~s) connected, connack: ~w, conn_attrs:~p~n", [ClientId, ConnAck, ConnAttrs]).

on_client_disconnected(ClientInfo = #{client_id := ClientId}, ReasonCode, _Env) ->
    {ok, ExchangeName} = application:get_env(?APP, hook_rabbitmq_exchangeName),
    Payload = [{action, <<"disconnected">>},
        {clientId, ClientId},
        {username, maps:get(username, ClientInfo)},
        {timestamp, calendar:datetime_to_gregorian_seconds(calendar:universal_time())-719528*24*3600}],
    emqx_plugin_rabbitmq_cli:publish(ExchangeName, Payload, <<"emqx_message">>),
    io:format("Client(~s) disconnected, reason_code: ~w~n", [ClientId, ReasonCode]).


%% Transform message and return
on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}, _Env) ->
    {ok, Message};

on_message_publish(Message = #message{topic = Topic, flags = #{retain := Retain}}, _Env) ->
    {ok, ExchangeName} = application:get_env(?APP, hook_rabbitmq_exchangeName),
    io:format("Public ExchangeName: (~s)~n",[ExchangeName]),
    io:format("Publish ~s~n", [emqx_message:format(Message)]),
    {ok, Payload} = format_payload(Message),
    emqx_plugin_rabbitmq_cli:publish(ExchangeName, Payload, <<"emqx_message">>),
    {ok, Message}.

%% Called when the plugin application stop
unload() ->
  
    emqx:unhook('client.connected', fun ?MODULE:on_client_connected/4),
    emqx:unhook('client.disconnected', fun ?MODULE:on_client_disconnected/3),
    emqx:unhook('message.publish', fun ?MODULE:on_message_publish/2).


format_payload(Message) ->
    Username = emqx_message:get_header(username, Message),
    Topic = Message#message.topic,
    MsgPayload = Message#message.payload,
    MsgPayload64 = list_to_binary(base64:encode_to_string(MsgPayload)),
    Payload = [{action, message_publish},
        {clientId, Message#message.from},
        {username, Username},
        {topic, Topic},
        {payload, MsgPayload64},
        {timestamp, emqx_time:now_secs(Message#message.timestamp)}],

    {ok, Payload}.
